package com.jianying.day08;

import com.jianying.beans.WaterSensor;
import com.jianying.func.WaterSensorMapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

/**
 * ClassName: Flink09_OperatorState_BroadcastState
 * Package: com.atguigu.day08
 * Description:
 *
 * @Author: tubage
 * @Create: 2024/4/9 15:31
 * @Version: 1.0
 */
public class Flink09_OperatorState_BroadcastState {
    public static void main(String[] args) throws Exception {
        // TODO 水位值超过指定阈值，发送告警，阈值可以动态修改
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // TODO 准备两条流
        SingleOutputStreamOperator<WaterSensor> wsDS = env.socketTextStream("localhost", 8888)
                .map(new WaterSensorMapFunction());

        SingleOutputStreamOperator<Integer> ruleDS = env.socketTextStream("localhost", 8889)
                .map(Integer::parseInt);

        // TODO 广播阈值流
        MapStateDescriptor<String, Integer> mapStateDescriptor =
                new MapStateDescriptor<String, Integer>("mapStateDescriptor",String.class,Integer.class);
        BroadcastStream<Integer> broadcastDS = ruleDS.broadcast(mapStateDescriptor);

        // TODO 两条流connect
        BroadcastConnectedStream<WaterSensor, Integer> connectDS = wsDS.connect(broadcastDS);

        // TODO 关联后数据处理
        connectDS.process(
                new BroadcastProcessFunction<WaterSensor, Integer, String>() {
                    @Override
                    public void processElement(WaterSensor value, ReadOnlyContext ctx, Collector<String> out) throws Exception {
                        // TODO 处理非广播流数据
                        ReadOnlyBroadcastState<String, Integer> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
                        Integer rule = broadcastState.get("rule");
                        Integer vc = value.getVc();
                        if(vc > rule) {
                            out.collect("当前水位值：" + vc + "超过阈值：" + rule);
                        }


                    }

                    @Override
                    public void processBroadcastElement(Integer value, Context ctx, Collector<String> out) throws Exception {
                        // TODO 处理广播流数据
                        // 获取广播状态
                        BroadcastState<String, Integer> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
                        // 将广播流中数据放到广播状态中
                        broadcastState.put("rule" ,value);

                    }
                }
        ).print();


        env.execute();

    }
}
